package original

import (
	"errors"
	"fmt"
	"miniDW/src/datapack"
	"miniDW/src/interfaces"
	"miniDW/src/log"
)

type ColumnFilterConfig struct {
	Columns []string          `json:"columns"`
	Rename  map[string]string `json:"rename"`
}

type OriginalColumnFilter struct {
	cfg         *ColumnFilterConfig
	tableStruct *datapack.TableStruct
	sourceName  string
	columnIndex []int
	count       int
	total       int
}

func (o *OriginalColumnFilter) Open(ctx interfaces.IContext) error {
	o.count = 0
	o.total = 0
	return nil
}

func (o *OriginalColumnFilter) Close() error {
	return nil
}

func (o *OriginalColumnFilter) SourceTableStructSetFinished(ctx interfaces.IContext) error {
	var err error
	log.Debug("ColumnFilter SourceTableStructSetFinished called")
	names := ctx.GetSourcesNames()
	if len(names) != 1 {
		return log.Error(fmt.Errorf("ColumnFilter not supports more than one Source."))
	}
	o.sourceName = names[0]
	sourceTableStruct, err := ctx.GetSourceTableStruct(o.sourceName)
	if err != nil {
		return err
	}

	o.columnIndex = make([]int, len(o.cfg.Columns))

	oldTs := ctx.GetTableStruct()
	newTs := datapack.NewTableStructByName(oldTs.Name())
	for i, column := range o.cfg.Columns {
		filed := sourceTableStruct.GetFieldTypeByName(column)
		if filed == nil {
			return log.Error(fmt.Errorf("Column '%s' has no field type", column))
		}
		newFiled := *filed
		if newName, ok := o.cfg.Rename[column]; ok {
			newFiled.Name = newName
		}
		o.columnIndex[i] = filed.Index
		err := newTs.AddFieldType(newFiled)
		if err != nil {
			return err
		}
	}
	err = ctx.SetTableStruct(newTs)
	if err != nil {
		return err
	}

	return nil
}

func (o *OriginalColumnFilter) Process(ctx interfaces.IContext) error {
	queue := ctx.GetRowQueue(o.sourceName)
	if queue == nil {
		return log.Error(fmt.Errorf("Query queue %s is nil.", o.sourceName))
	}

	newRow := make(datapack.Row, len(o.columnIndex))

	rowCount := queue.Len()
	if rowCount <= 0 && queue.IsEOF() {
		log.Debug("ColumnFilter Row queue is empty.")
		err := ctx.SendRowData(nil)
		if err != nil {
			return err
		}
		ctx.Stop()
		return nil
	}
	for rowCount > 0 {
		row := queue.Dequeue()
		for i, columnIndex := range o.columnIndex {
			newRow[i] = (*row)[columnIndex]
		}

		err := ctx.SendRowData(&newRow)
		if err != nil {
			return err
		}

		rowCount--
	}

	return nil
}

func parseConfig(params interface{}) (*ColumnFilterConfig, error) {
	cfg := &ColumnFilterConfig{}
	if m, ok := params.(map[string]interface{}); ok {
		if columns, ok := m["columns"].([]interface{}); ok {
			cfg.Columns = make([]string, len(columns))
			for i, column := range columns {
				if str, ok := column.(string); ok {
					cfg.Columns[i] = str
				}
			}
		}
		if renames, ok := m["rename"].(map[string]interface{}); ok {
			cfg.Rename = make(map[string]string)
			for k, v := range renames {
				cfg.Rename[k] = v.(string)
			}
		}
		return cfg, nil
	} else {
		return nil, log.Error(errors.New("data is not map[string]interface{}"))
	}
}

func CreateOriginalColumnFilter(params map[string]interface{}) (interfaces.IComponentFunction, error) {
	var err error
	filter := &OriginalColumnFilter{}
	filter.cfg, err = parseConfig(params)
	if err != nil {
		return nil, err
	}
	filter.tableStruct = &datapack.TableStruct{}
	return filter, nil
}
